package com.smasher.mqtt.config;

import android.os.Handler;
import android.os.Looper;
import android.os.Message;
import android.util.Log;

import androidx.annotation.NonNull;

import com.smasher.mqtt.interfaces.MessageReceiver;
import com.smasher.mqtt.receivers.TopicMessageReceiver;

import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageDispatcher implements Handler.Callback {

    private final Map<String, List<MessageReceiver>> receivers = new ConcurrentHashMap<>();

    private final BlockingQueue<MessageEvent> messageQueue = new LinkedBlockingQueue<>(100);
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    private final Handler handler = new Handler(Looper.getMainLooper(), this);

    /**
     * 是否停止线程
     */
    private boolean stopThread;

    private static final int MSG_DISPATCH = 0;

    public static final String TAG = "MessageDispatcher";

    public MessageDispatcher() {
        // 启动工作线程
        stopThread = false;
        executor.submit(new Worker());
    }

    public void addReceiver(MessageReceiver receiver) {
        if (receiver == null) return;

        String topic = "default";
        if (receiver instanceof TopicMessageReceiver) {
            String customTopic = ((TopicMessageReceiver) receiver).getTopic();
            if (customTopic != null) {
                topic = customTopic;
            }
        }

        synchronized (receivers) {
            List<MessageReceiver> list = receivers.get(topic);
            if (list == null) {
                list = new CopyOnWriteArrayList<>();
                receivers.put(topic, list);
            }
            list.add(receiver);
        }
    }

    public void removeReceiver(MessageReceiver receiver) {
        if (receiver == null) return;

        String topic = "default";
        if (receiver instanceof TopicMessageReceiver) {
            String customTopic = ((TopicMessageReceiver) receiver).getTopic();
            if (customTopic != null) {
                topic = customTopic;
            }
        }

        synchronized (receivers) {
            List<MessageReceiver> list = receivers.get(topic);
            if (list != null) {
                list.remove(receiver);
                if (list.isEmpty()) {
                    receivers.remove(topic);
                }
            }
        }
    }

    public void dispatch(String topic, MqttMessage message) {
        String payload = new String(message.getPayload());
        boolean isOffered = messageQueue.offer(new MessageEvent(topic, payload));
        Log.d(TAG, "dispatch: isOffered=" + isOffered);
    }

    @Override
    public boolean handleMessage(@NonNull Message msg) {
        if (msg.what == MSG_DISPATCH) {
            MessageEvent event = (MessageEvent) msg.obj;
            if (event != null) {
                String topic = event.topic;
                String message = event.payload;
                dispatchMessage(topic, message);
            }

            return true;
        }
        return false;
    }


    private void dispatchMessage(String topic, String message) {
        List<MessageReceiver> targetList = receivers.get(topic);
        if (targetList != null) {
            for (MessageReceiver receiver : targetList) {
                receiver.onMessageReceived(message);
            }
        }
    }

    private class Worker implements Runnable {
        @Override
        public void run() {
            while (!stopThread) {
                try {
                    MessageEvent event = messageQueue.take();
                    Message message = Message.obtain();
                    message.what = MSG_DISPATCH;
                    message.obj = event;
                    handler.sendMessage(message);
                } catch (Exception e) {
                    Log.e(TAG, "run: ", e);
                }
            }
        }
    }

    public void release() {
        stopThread = true;
        messageQueue.clear();
        handler.removeCallbacksAndMessages(null);
        receivers.clear();
        executor.shutdown();
    }

    private static class MessageEvent {
        public final String topic;
        public final String payload;

        public MessageEvent(String topic, String payload) {
            this.topic = topic;
            this.payload = payload;
        }
    }
}
